/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package org.xeerkat.xmpp.server;

import java.util.ArrayList;
import java.util.List;
import java.util.logging.Logger;

/**
 *
 * @author alex
 */
public class ThreadTaskPool {
   class Worker {
      Thread thread;
      Runnable task;
      Worker() {
         Runnable r = new Runnable() {
            public void run() {
               while (running) {
                  synchronized (ThreadTaskPool.this) {
                     idleCount++;
                  }
                  task = next();
                  if (running) {
                     synchronized (ThreadTaskPool.this) {
                        idleCount--;
                     }
                     task.run();
                     task = null;
                  }
               }
            }
         };
         thread = new Thread(group,r,group.getName()+" "+group.activeCount()+1);
         thread.start();
      }
   }

   int minimum;
   ThreadGroup group;
   List<Worker> workers;
   int idleCount;
   List<Runnable> work;
   boolean running;
   Logger log;

   public ThreadTaskPool(Logger log,ThreadGroup group,int minimum) {
      this.log = log;
      this.minimum = minimum;
      this.group = group;
      this.workers = new ArrayList<Worker>();
      this.work = new ArrayList<Runnable>();
      this.running = true;
      this.idleCount = 0;
   }
   protected Runnable next() {
      synchronized (work) {
         if (running && work.size()==0) {
            try {
               work.wait();
            } catch (InterruptedException ex) {
            }
         }
         return (running && work.size()>0) ? work.remove(0) : null;
      }
   }
   
   public void addTask(Runnable r)
   {
      synchronized (work) {
         work.add(r);
         if (idleCount==0) {
            workers.add(new Worker());
         }
         work.notify();
      }
   }
   
   public void start() {
      for (int i=0; i<minimum; i++) {
         workers.add(new Worker());
      }
   }
   
   public void stop() {
      running = false;
      synchronized (work) {
         work.notifyAll();
      }
      join();
   }
   
   protected void join() {
      log.info("Active threads in "+group.getName()+": "+group.activeCount());
      for (Worker worker : workers) {
         if (worker.thread.isAlive()) {
            if (worker.task!=null && worker.task instanceof Stoppable) {
               log.info("Requesting thread "+worker.thread.getName()+" to stop.");
               ((Stoppable)worker.task).stop();
            }
         }
      }
      
      for (Worker worker : workers) {
         if (worker.thread.isAlive()) {
            log.info("Waiting on thread "+worker.thread.getName());
            try {
               worker.thread.join(500);
            } catch (InterruptedException ex) {
               ex.printStackTrace();
            }
            if (worker.thread.isAlive()) {
               log.severe("Thread "+worker.thread.getName()+" didn't join!");
            }
            if (worker.thread.isAlive() && worker.task!=null && worker.task instanceof Stoppable) {
               worker.thread.interrupt();
               //((Stoppable)worker.task).kill();
               log.info("Waiting on thread "+worker.thread.getName());
               try {
                  worker.thread.join(500);
               } catch (InterruptedException ex) {
                  ex.printStackTrace();
               }
               if (worker.thread.isAlive()) {
                  log.severe("Thread "+worker.thread.getName()+" still didn't join!");
               }
            }
         }
      }
      if (group.activeCount()>0) {
         log.severe("Not all threads in "+group.getName()+" joined. "+group.activeCount()+" threads are still active.");
      } else {
         log.info("All treads for "+group.getName()+" are stopped.");
      }
   }
}
